package com.jun.admin.core.listener;

import com.jun.admin.core.model.constant.RedisKeyConstant;
import com.jun.common.redis.stream.RedisStreamConsumer;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * @author yuejun.li
 * @date 2024/8/13 22:26
 * redis的消息队列消费者
 */
@Component
@Slf4j
public class RedisStreamMessageListener extends RedisStreamConsumer {

    private RedisTemplate redisTemplate;

    public RedisStreamMessageListener(RedisTemplate redisTemplate,StreamMessageListenerContainer streamMessageListenerContainer) {
        super(streamMessageListenerContainer, redisTemplate);
        this.redisTemplate = redisTemplate;
        Consumer consumer = Consumer.from(RedisKeyConstant.REDIS_STREAM_CONSUMER_GROUP, RedisKeyConstant.REDIS_STREAM_CONSUMER_1);
        receive(consumer, StreamOffset.create(RedisKeyConstant.REDIS_STREAM_QUEUE, ReadOffset.lastConsumed()),true);
    }

    @Override
    protected void onStreamMessage(MapRecord<String, String, String> message) {
        String stream = message.getStream();
        RecordId id = message.getId();
        Map<String, String> map = message.getValue();
        log.info("[自动ACK]接收到一个消息 stream:[{}],id:[{}],value:[{}]", stream, id, map);
        redisTemplate.opsForStream().delete(RedisKeyConstant.REDIS_STREAM_QUEUE, id.getValue());
    }
}
